%%%====================================================================
%%% File    : ptool_pipe_keyfold.erl
%%% Author  :  <vjache>
%%% Description: TODO Describe.
%%%
%%% Created : 30.03.2009
%%%====================================================================

-module(ptool_pipe_keyfold).

%%
%% Include files
%%
-include("../include/pipe.hrl").
-include("../include/iter.hrl").
%%
%% Exported Functions
%%
-export([
		 new/2,
		 open/1,
		 push/2,
		 flush/1,
		 close/1,
		 test/1,
		 test/0
		]).

%%====================================================================
%% API Functions
%%====================================================================


new({_PToolDescr,_FoldFun,_FinishFun,_Acc0}=Arg,NextPipe) ->
	pipe:pipe_from_module(?MODULE, Arg, NextPipe).

open({{PToolDescr,FoldFun,FinishFun,Acc0}=_Param,NextPipe}) -> %%PipeState
	PToolState=ptool:open(PToolDescr),
	PipeState=pipe:open(NextPipe),
	{PToolState,PipeState,FoldFun,FinishFun,Acc0}.

push({_PToolState,_PipeState,_FoldFun,_FinishFun,_Acc0}=S,?EOF) ->
	S;
push({PToolState,_PipeState,FoldFun,FinishFun,Acc0}=S,Item) ->
    AccHandle=ptool:obtain_acc(PToolState,Item,FoldFun,FinishFun,Acc0),
	ok=ptool:send_acc(PToolState,AccHandle, Item),
	S.
flush({PToolState,PipeState,FoldFun,FinishFun,_Acc0}=_S) ->
	Promises=ptool:finish_all(PToolState),
	NextPipeStateNew=ptool:yield_atwill_all(PToolState, Promises,
						   fun(Acc,PipeSt)->
								   pipe:push(PipeSt,Acc)
						   end,PipeState),
	{PToolState,NextPipeStateNew,FoldFun,FinishFun,_Acc0}.
close({PToolState,PipeState,_FoldFun,_FinishFun,_Acc0}=_S) ->
	ok=pipe:close(PipeState),
	ok=ptool:close(PToolState).


%%====================================================================
%% Test Functions
%%====================================================================

test() ->
	List=[
		  {predicate,"pred","subj","obj","graph"},
		  {predicate,"pred","subj1","obj","graph"},
		  {predicate,"pred","subj2","obj--","graph"},
		  {predicate,"pred","subj3","obj74","graph"},
		  {predicate,"pred","subj4","obj45","graph"},
		  {predicate,"pred","subj5","obj%%","graph"},
		  {predicate,"pred","subj6","obj","graph"},
		  {predicate,"pred1","subj","obj","graph"},
		  {predicate,"pred1","subj1","obj","graph"},
		  {predicate,"pred1","subj2","obj--","graph"},
		  {predicate,"pred2","subj3","obj74","graph"},
		  {predicate,"pred2","subj4","obj45","graph"},
		  {predicate,"pred2","subj5","obj%%","graph"},
		  {predicate,"pred2","subj6","obj","graph"},
		  {predicate,"pred","subj","obj","graph"},
		  {predicate,"pred","subj1","obj","graph"},
		  {predicate,"pred","subj2","obj--","graph"},
		  {predicate,"pred","subj3","obj74","graph"},
		  {predicate,"pred","subj4","obj45","graph"},
		  {predicate,"pred","subj5","obj%%","graph"},
		  {predicate,"pred","subj6","obj","graph"},
		  {predicate,"pred1","subj","obj","graph"},
		  {predicate,"pred1","subj1","obj","graph"},
		  {predicate,"pred1","subj2","obj--","graph"},
		  {predicate,"pred2","subj3","obj74","graph"},
		  {predicate,"pred2","subj4","obj45","graph"},
		  {predicate,"pred2","subj5","obj%%","graph"},
		  {predicate,"pred2","subj6","obj","graph"},
		  {predicate,"pred","subj","obj","graph"},
		  {predicate,"pred","subj1","obj","graph"},
		  {predicate,"pred","subj2","obj--","graph"},
		  {predicate,"pred","subj3","obj74","graph"},
		  {predicate,"pred","subj4","obj45","graph"},
		  {predicate,"pred","subj5","obj%%","graph"},
		  {predicate,"pred","subj6","obj","graph"},
		  {predicate,"pred1","subj","obj","graph"},
		  {predicate,"pred1","subj1","obj","graph"},
		  {predicate,"pred1","subj2","obj--","graph"},
		  {predicate,"pred2","subj3","obj74","graph"},
		  {predicate,"pred2","subj4","obj45","graph"},
		  {predicate,"pred2","subj5","obj%%","graph"},
		  {predicate,"pred2","subj6","obj","graph"},
		  {predicate,"pred","subj","obj","graph"},
		  {predicate,"pred","subj1","obj","graph"},
		  {predicate,"pred","subj2","obj--","graph"},
		  {predicate,"pred","subj3","obj74","graph"},
		  {predicate,"pred","subj4","obj45","graph"},
		  {predicate,"pred","subj5","obj%%","graph"},
		  {predicate,"pred","subj6","obj","graph"},
		  {predicate,"pred1","subj","obj","graph"},
		  {predicate,"pred1","subj1","obj","graph"},
		  {predicate,"pred1","subj2","obj--","graph"},
		  {predicate,"pred2","subj3","obj74","graph"},
		  {predicate,"pred2","subj4","obj45","graph"},
		  {predicate,"pred2","subj5","obj%%","graph"},
		  {predicate,"pred2","subj6","obj","graph"},
		  {predicate,"pred3","subj","obj","graph"},
		  {predicate,"pred3","subj1","obj","graph"},
		  {predicate,"pred3","subj2","obj--","graph"},
		  {predicate,"pred3","subj3","obj74","graph"},
		  {predicate,"pred3","subj4","obj45","graph"},
		  {predicate,"pred3","subj5","obj%%","graph"},
		  {predicate,"pred3","subj6","obj","graph"},
		  {predicate,"pred31","subj","obj","graph"},
		  {predicate,"pred31","subj1","obj","graph"},
		  {predicate,"pred31","subj2","obj--","graph"},
		  {predicate,"pred32","subj3","obj74","graph"},
		  {predicate,"pred32","subj4","obj45","graph"},
		  {predicate,"pred32","subj5","obj%%","graph"},
		  {predicate,"pred23","subj6","obj","graph"}
		 ],
	test(List).

test(L) ->
	S=pipe:open(new({ptool_local:new(_KeyFun=fun(Item)->
													 element(2,Item)
											 end),
					 _FoldFun=fun(Key,Item,Acc)->
									  Acc+1
							  end,
					 _FinishFun=fun(Key,Acc)->
										{Key,Acc}
								end,
					 _Acc0=0},pipe_console:new())),
	{IterState,PipeState}=pipe:push_each(S, iter:from_list(L)),
	iter:close(IterState),
	pipe:close(PipeState).


%%====================================================================
%% Local Functions
%%====================================================================